package com.intct.common;

import org.apache.commons.lang3.StringUtils;

/**
 * @author gufg
 * @since 2025-10-16 11:03
 */
public class FlinkSqlWithUtil {
    /**
     *
     * @param db
     * @param table
     * @param startMode
     * @return
     */
    public static String getMysqlCdcWith(String db, String table, String startMode) {
        return " WITH (                                          " +
                  "  'connector' = 'mysql-cdc'                   " +
                  "     ,'hostname' = 'cdh-node'                 " +
                  "     ,'port' = '13306'                        " +
                  "     ,'username' = 'root'                     " +
                  "     ,'password' = 'Test_090110'              " +
                  "     ,'database-name' = '" + db + "'          " +
                  "     ,'table-name' = '" + table + "'          " +
                  "     ,'scan.startup.mode' = '"+startMode+"'   " +
                  " ) ";
    }

    /**
     *
     * @param ns
     * @param table
     * @return
     */
    public static String getHBaseSinkWith(String ns, String table) {
        return " WITH (                                          " +
                "  'connector' = 'hbase-2.2'                     " +
                "     ,'table-name' = '"+ ns + ":" + table+"'    " +
                "     ,'zookeeper.quorum'    = 'cdh-node:2181'   " +
                " ) ";
    }

    /**
     *
     * @param ns
     * @param table
     * @param asyncEnabled
     * @return
     */
    public static String getHBaseLookUpWith(String ns, String table, boolean asyncEnabled) {
        return " WITH (                                                 " +
                "  'connector' = 'hbase-2.2'                            " +
                "     ,'table-name' = '"+ ns + ":" + table+"'           " +
                "     ,'zookeeper.quorum'    = 'cdh-node:2181'          " +
                "     ,'lookup.async'        = ' "+ asyncEnabled +" '   " +
                " ) ";
    }

    /**
     *
     * @param topic
     * @return
     */
    public static String getKafkaSinkWith(String topic) {
        return  " WITH (                                                 " +
                "  'connector' = 'upsert-kafka'," +
                "  'topic' = ' " + topic + " '," +
                "  'properties.bootstrap.servers' = 'cdh-node:9092'," +
                "  'key.format' = 'json'," +
                "  'value.format' = 'json'" +
                " ) ";
    }

    /**
     *
     * @param topic
     * @param groupId
     * @return
     */
    public static String getKafkaSourceWith(String topic, String groupId, String clientId) {
        return  " WITH (                                              " +
                "    'connector'='kafka'                               " +
                "    ,'topic'=' "+ topic +" '                          " +
                "    ,'properties.bootstrap.servers'='cdh-node:9092'   " +
                "    ,'properties.group.id'=' " + groupId + " '        " +
                "    ,'properties.enable.metrics'='false'              " +
                "    ,'properties.client.id' = '"+ clientId +"'        " +
                "    ,'properties.metric.reporters' = ''               " +
                "    ,'properties.auto.offset.reset' = 'earliest'      " +
                "    ,'format'='json'                                  " +
                "    ,'scan.startup.mode'='latest-offset'            " +
                "    ,'json.ignore-parse-errors' = 'true'              " +
                ")                                                     ";
    }

    /**
     *
     * @param ttl
     * @param command
     * @return
     */
    public static String getRedisSourceWith(String ttl, String command) {
        return  " WITH (                                         " +
                "  'connector'='redis'                            " +
                "  ,'host'='cdh-node'                             " +
                "  ,'port'='16379'                                " +
                "  ,'redis-mode'='single'                         " +
                "  ,'password'='Test_090110'                      " +
                "  ,'ttl'='" + ttl + "'                           " +
                "  ,'command'='" + command + "'                   " +
                ")                                                ";
    }
}
